package com.example.springboot.controller;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.Arrays;
import java.util.Map;

/**
 * @Author:ChenZhangKun
 * @Date: 2021/5/25 17:05
 */
//@RestController
@RequestMapping("/kafka")
public class KafkaController {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("/send")
    public String sendKafka(@RequestParam("message") String message) {
        // 加入自己的拦截器
        Map<String, Object> configurationProperties = kafkaTemplate.getProducerFactory().getConfigurationProperties();
        configurationProperties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, Arrays.asList(
                "com.example.springboot.config.kafka.TimerInterceptor",
                "com.example.springboot.config.kafka.CounterInterceptor"
        ));
        kafkaTemplate.send("first", message);
        return "发送成功";
    }

    /**
     * 发送自己配置的topic
     *
     * @param message
     * @return
     */
    @GetMapping("/send2")
    public String sendKafka_2(@RequestParam("message") String message) {
        // 发送配置的topic
        kafkaTemplate.send("second", message);
        return "发送成功";
    }
}
